package com.bbx.flink.demo.ProcessFunctionAllDemo;

import com.esotericsoftware.minlog.Log;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.*;


/**
 * 需求   数值 10s 连续上升，输出报警和数值
 */
public class ContinuousUpDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1).socketTextStream("192.168.10.131", 10003)
                .map(i -> Integer.parseInt(i))
                .keyBy(i -> i % 2 == 0)
                .process(new ContinuousUpFunction())
                .print();
        env.execute();
    }


}

@Slf4j
class ContinuousUpFunction extends KeyedProcessFunction<Boolean, Integer, Map<String, Object>> {

    private MapState<String, Object> mapState;

    @Override
    public void open(Configuration parameters) {
        mapState = getRuntimeContext().getMapState(
                new MapStateDescriptor<String, Object>("startTime", String.class, Object.class));
    }


    @Override
    public void processElement(Integer integer, Context context, Collector<Map<String, Object>> collector) throws Exception {
        //获取状态内的值
        if (mapState.contains("startTime")) {
            Long startTime = (Long) mapState.get("startTime");
            Object valueList = mapState.get("valueList");
            if (valueList != null) {
                List valueListTmp = (List) valueList;
                int size = valueListTmp.size();
                Integer oldValue = (Integer) valueListTmp.get(size - 1);
                if (integer > oldValue) {
                    valueListTmp.add(integer);
                    Log.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
                    mapState.put("valueList", valueListTmp);
                } else {
                    //清空状态
                    mapState.clear();
                    //清空定时任务
                    context.timerService().deleteProcessingTimeTimer(startTime + 10000L);
                }
            }
        }
        /**
        * 此处不能使用else 因为 当传入的数据不能满足条件，需要将当前传入的数据写入
        */
        if (!mapState.contains("startTime")) {
            Long timestamp = context.timerService().currentProcessingTime();
            log.info("~~~~~~~~~~~~~~~timestamp[{}]", timestamp);
            List<Integer> valueList = new ArrayList<>();
            valueList.add(integer);
            mapState.put("startTime", timestamp);
            mapState.put("valueList", valueList);
            //注册定时
            context.timerService().registerProcessingTimeTimer(timestamp + 10000L);
        }
    }

    /*
     *定时器
     */
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Map<String, Object>> out) throws Exception {
        Log.info("进入定时器");
        Object startTime = mapState.get("startTime");
        Object valueList = mapState.get("valueList");
        Map<String, Object> map = new HashMap<>();
        map.put("valueList", valueList);
        map.put("startTime", startTime);
        //清空state
        mapState.clear();
        out.collect(map);
    }


    @Override
    public void close() throws Exception {
        mapState.clear();
    }
}